package org.example.queueadmin.task;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.example.queueadmin.dlq.DeadLetterService;
import org.example.queueadmin.log.MessageLogService;
import org.example.queueadmin.task.runner.KafkaToRabbitRunner;
import org.example.queueadmin.task.runner.RabbitToKafkaRunner;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;
import java.util.List;

@Slf4j
@Service
@RequiredArgsConstructor
public class TaskService {
  private final TransferTaskRepository repository;
  private final TaskManager taskManager;
  private final DeadLetterService dlq;
  private final MessageLogService messageLogService;

  public List<TransferTask> list() { return repository.findAll(); }

  @Transactional
  public TransferTask createOrUpdate(TransferTask t) {
    // 如果同名任务存在，则更新（幂等创建）
    if (t.getId() == null) {
      repository.findByName(t.getName()).ifPresent(existing -> {
        t.setId(existing.getId());
        // 保留运行时状态字段
        t.setStatus(existing.getStatus());
        t.setProcessed(existing.getProcessed());
        t.setFailed(existing.getFailed());
        t.setLastStartedAt(existing.getLastStartedAt());
        t.setLastStoppedAt(existing.getLastStoppedAt());
        t.setLastErrorAt(existing.getLastErrorAt());
        t.setLastErrorMsg(existing.getLastErrorMsg());
      });
    }
    if (t.getId() == null) {
      t.setStatus("STOPPED");
      t.setProcessed(0L);
      t.setFailed(0L);
      t.setLastStartedAt(null);
      t.setLastStoppedAt(null);
      t.setLastErrorAt(null);
      t.setLastErrorMsg(null);
    }
    return repository.save(t);
  }

  public void delete(Long id) {
    stop(id);
    repository.deleteById(id);
  }

  public void start(Long id) {
    TransferTask t = repository.findById(id).orElseThrow();
    TaskRunner existing = taskManager.get(id);
    if (existing != null) { existing.stop(); }

    TaskRunner runner = switch (t.getDirection()) {
      case KAFKA_TO_RABBIT -> new KafkaToRabbitRunner(t, dlq, repository, messageLogService);
      case RABBIT_TO_KAFKA -> new RabbitToKafkaRunner(t, dlq, repository, messageLogService);
    };
    taskManager.register(id, runner);

    try {
      runner.start();
      t.setStatus("RUNNING");
      t.setLastStartedAt(Instant.now());
      repository.save(t);
    } catch (Exception e) {
      log.error("Failed to start task {}: {}", t.getName(), e.getMessage(), e);
      t.setStatus("ERROR");
      t.setFailed((t.getFailed() == null ? 0 : t.getFailed()) + 1);
      t.setLastErrorAt(Instant.now());
      t.setLastErrorMsg(e.getMessage());
      repository.save(t);
      taskManager.unregister(id);
      throw e;
    }
  }

  public void stop(Long id) {
    TransferTask t = repository.findById(id).orElse(null);
    if (t != null) {
      // 立即更新状态为停止中
      t.setStatus("STOPPING");
      repository.save(t);
    }

    TaskRunner r = taskManager.get(id);
    if (r != null) {
      try {
        r.stop();
      } catch (Exception e) {
        log.error("Error stopping task {}: {}", id, e.getMessage(), e);
      }
    }

    // 最终更新为已停止
    if (t != null) {
      t.setStatus("STOPPED");
      t.setLastStoppedAt(Instant.now());
      repository.save(t);
    }

    taskManager.unregister(id);
  }

  public void restart(Long id) { stop(id); start(id); }

  public String status(Long id) {
    TaskRunner r = taskManager.get(id);
    if (r == null) {
      TransferTask t = repository.findById(id).orElseThrow();
      return t.getStatus();
    }
    return r.status();
  }

  @Transactional
  public void clearError(Long id) {
    TransferTask t = repository.findById(id).orElseThrow();
    t.setFailed(0L);
    t.setLastErrorAt(null);
    t.setLastErrorMsg(null);
    repository.save(t);
  }

  public TransferTask findById(Long id) {
    return repository.findById(id).orElseThrow();
  }
}

